package com.ds.infrastructure.hyperspace.container.endpoint.destination;

import com.ds.infrastructure.hyperspace.common.EndpointProtocolEnum;
import com.ds.infrastructure.hyperspace.container.endpoint.connector.HyperspaceConnector;
import com.ds.infrastructure.hyperspace.container.endpoint.connector.NettyEndpointNodeStream;
import com.ds.infrastructure.hyperspace.container.flow.context.HyperspaceDuplexFlowContext;
import com.ds.infrastructure.hyperspace.container.flow.operation.TargetFlowOperation;
import com.ds.infrastructure.hyperspace.container.internal.NettyContainer;
import com.ds.infrastructure.hyperspace.container.internal.util.AddressesUtil;
import io.netty.channel.Channel;
import io.netty.channel.ChannelDuplexHandler;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelId;
import lombok.extern.slf4j.Slf4j;

import static com.ds.infrastructure.hyperspace.container.internal.util.StringConstant.HYPHEN;
import static io.netty.channel.ChannelFutureListener.FIRE_EXCEPTION_ON_FAILURE;

/**
 * @author  Weishaoying
 */
@Slf4j
public abstract class AbstractTargetProtocolAdapter extends ChannelDuplexHandler implements TargetProtocolAdapter {

    protected final HyperspaceConnector hyperspaceConnector;

    protected final TargetFlowOperation<HyperspaceDuplexFlowContext> targetFlowOperation;

    protected final TargetEndpointNode targetEndpointNode;

    protected Channel channel;

    protected ChannelHandlerContext channelHandlerContext;

    protected NettyEndpointNodeStream stream;

    private boolean isOpenChannel = true;

    private boolean fireChannelInactive = false;

    protected String targetRemoteAddress = HYPHEN;

    public AbstractTargetProtocolAdapter(NettyContainer nettyContainer, HyperspaceConnector hyperspaceConnector,
                                         TargetEndpointNode targetEndpointNode) {
        this.hyperspaceConnector = hyperspaceConnector;
        this.targetFlowOperation = nettyContainer.getFlowPipelineOrchestrator().getTargetFlowOperation();
        this.targetEndpointNode = targetEndpointNode;
    }

    @Override
    public void channelActive(ChannelHandlerContext ctx) throws Exception {
        this.channelHandlerContext = ctx;
        this.channel = ctx.channel();
        this.targetRemoteAddress = AddressesUtil.getIpAndPort(this.channel.remoteAddress());
        targetFlowOperation.handleChannelActive(targetEndpointNode, stream, this);
    }

    @Override
    public void channelInactive(ChannelHandlerContext ctx) throws Exception {
        if (!fireChannelInactive && stream != null) {
            fireChannelInactive = true;
            targetFlowOperation.handleChannelInactive(targetEndpointNode,
                    stream, this, connectionDuplexFlowContext());
        }
    }

    @Override
    public void handlerRemoved(ChannelHandlerContext ctx) throws Exception {
        channelInactive(ctx);
    }

    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
        try {
            handleChannelRead(ctx, msg);
        } catch (Exception ex) {
            releaseData(msg);
            throw ex;
        }
    }

    /**
     * Is called for each message
     *
     * @param ctx
     * @param msg
     * @throws Exception
     */
    protected abstract void handleChannelRead(ChannelHandlerContext ctx, Object msg);

    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
        String errorMessage = String.format("%s-Occur an exception that target endpoint handler unexpected,node info[%s]," +
                "target address %s!cause by %s", getChannelId(), targetEndpointNode.getKey(), targetRemoteAddress, cause);
        log.error(errorMessage, cause);
        targetFlowOperation.handleExceptionCaught(targetEndpointNode.getEndpoint().getNamespace(), cause);
        closeTargetChannel();
    }

    private ChannelId getChannelId() {
        if (channel == null) {
            return null;
        }
        return channel.id();
    }

    @Override
    public void attachStream(NettyEndpointNodeStream stream) {
        this.stream = stream;
        this.stream.setTargetProtocolAdapter(this);
    }

    @Override
    public void writeToTarget(HyperspaceDuplexFlowContext duplexFlowContext, Object data) {
        stream.touch();
        if (!requireDuplexFlowContext(duplexFlowContext)) {
            return;
        }
        writeDataToTarget(duplexFlowContext, data);
        targetFlowOperation.handleInput(duplexFlowContext, data);
    }

    @Override
    public void flushToTarget(HyperspaceDuplexFlowContext duplexFlowContext) {
        channelHandlerContext.flush();
        targetFlowOperation.handleInputComplete(duplexFlowContext);
    }

    @Override
    public boolean isChannelActive() {
        return channel != null && channel.isActive();
    }

    @Override
    public void closeTargetChannel() {
        if (isOpenChannel && channel.isOpen()) {
            isOpenChannel = false;
            channel.flush().close().addListener(FIRE_EXCEPTION_ON_FAILURE);
        }
    }

    @Override
    public void checkTimeoutDuplexFlowContext() {
        // for sub class implements
    }

    @Override
    public void changeTargetChannelReadable(boolean readable) {
        channel.config().setAutoRead(readable);
    }

    @Override
    public void switchProtocol(HyperspaceDuplexFlowContext duplexFlowContext,
                               EndpointProtocolEnum protocolCluster) {
        // no implements
    }

    @Override
    public void resetChannelActiveState(ChannelHandlerContext channelHandlerContext,
                                        NettyEndpointNodeStream stream, HyperspaceDuplexFlowContext duplexFlowContext) {
        this.channel = channelHandlerContext.channel();
        this.channelHandlerContext = channelHandlerContext;
        attachStream(stream);
    }

    /**
     * Write data to target channel
     *
     * @param duplexFlowContext
     * @param data
     */
    public abstract void writeDataToTarget(HyperspaceDuplexFlowContext duplexFlowContext, Object data);

}
